package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwdTrafficSplitLogApp {


    //1 、 清洗（看是否能正常转为json ，否则侧输出到另外的脏数据主题中）
    //2、  is_new 作为前端埋点留下的新老用户标记
    //        什么情况不对 ：
    //              前端把数据可以持久化到cookie，只要是访问同一个域名，再次打开浏览器依然可以读取上次的cookie。手机端还有缓存。
    //           cookie和缓存，有可能被清理，手动清理，包括过期时间。重装系统等等。
    //3、 数据分流
    //       目前所有的日志数据，全部都耦合在了一个主题中 topic_log
    //        页面访问主题、启动程序主题、报错主题、曝光主题、页面动作主题。
    public static void main(String[] args) throws Exception {
        //1  环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        String topic = "topic_log";
        String groupId = "dwd_traffic_split_log";

        // 相关的配置
        // 并行度、 状态后端  hashmap rocksdb  、 checkpoint 、重试参数 、调优参数

        // 2  来源
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStream = env.addSource(kafkaConsumer);

        //3 清洗（看是否能正常转为json ，否则侧输出到另外的脏数据主题中）
        //   1  map  2 richmap 3 filter 4 richfilter  5 process
        OutputTag<String> dirtyDataTag = new OutputTag<String>("dirty_data") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String json, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(json);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(dirtyDataTag, json);  //转换不成功的数据，打上脏数据标签
                }
            }
        });

        DataStream<String> dirtyDataStream = jsonObjStream.getSideOutput(dirtyDataTag);

        //2、  is_new 作为前端埋点留下的新老用户标记 ,可能不准确,要通过实时计算进行修复。
        //  通过一个状态 来记录识别用户的标识 ，用户的首次访问时间(状态存储）

        //


        //直接用一个 MapState 的算子状态 key用mid  value 用lastdate 可以吗？
        // 或者 用 valueState 的 键控状态   value 用lastdate 可以吗？
        //
        // 使用keyby

        KeyedStream<JSONObject, String> midKeyedStream = jsonObjStream.keyBy(jsoObj -> jsoObj.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> jsonObjAfterIsnewCheckStream = midKeyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {

            //状态描述符
            ValueStateDescriptor<String> firstVisitDateStateDesc = new ValueStateDescriptor<String>("first_visit_date", String.class);
            //声明
            ValueState<String> firstVisitDateState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDateState = getRuntimeContext().getState(firstVisitDateStateDesc);
            }

            //    如果前端is_new 1    要考虑是否正确
            //       【本次用户访问记录的时间戳】和  状态中的【用户的首次访问时间】 作比较
            ///      情况1：  【用户的首次访问时间】 不存在       is_new 保持为1 , 把 【本次用户访问记录的时间戳】 写入 【用户的首次访问时间】
            //       情况2：  【本次用户访问记录的时间戳】 大于【用户的首次访问时间】 时 ，说明用户曾经访问过 is_new 设为0
            //       情况3：  【本次用户访问记录的时间戳】 =【用户的首次访问时间】  is_new 保持为1
            //
            //     如果前端is_new 0   不用修复
            //     【用户的首次访问时间】 不存在
            //       补充 【用户的首次访问时间】 ，放前一日的日期
            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                Long ts = jsonObject.getLong("ts");
                String curDateStr = DateFormatUtil.toDate(ts);
                String firstDateStr = firstVisitDateState.value();
                if ("1".equals(isNew)) {
                    //    如果前端is_new 1    要考虑是否正确
                    //       【本次用户访问记录的时间戳】和  状态中的【用户的首次访问日期】 作比较
                    if (firstDateStr == null || firstDateStr.length() == 0) {
                        ///      情况1：  【用户的首次访问日期】 不存在         is_new 保持为1 , 把 【本次用户访问记录的时间戳】 写入 【用户的首次访问时间】
                        firstVisitDateState.update(curDateStr);
                    } else if (!firstDateStr.equals(curDateStr)) {
                        //       情况2：  【本次用户访问记录的时间戳】 大于【用户的首次访问日期】 时 ，说明用户曾经访问过 is_new 设为0
                        jsonObject.getJSONObject("common").put("is_new", "0");
                    }
                    //       情况3：  【本次用户访问记录的时间戳】 =【用户的首次访问日期】  is_new 保持为1
                    //         不做任何处理
                    //
                } else {
                    //     如果前端is_new 0   不用修复
                    //     【用户的首次访问时间】 不存在
                    //       补充 【用户的首次访问时间】 ，放前一日的日期
                    Long yesterdayTs = ts - (1000 * 3600 * 24);
                    String yesterdayDt = DateFormatUtil.toDate(yesterdayTs);
                    firstVisitDateState.update(yesterdayDt);
                }
                return jsonObject;
            }
        });


        //3  打印
        // jsonObjAfterIsnewCheckStream.print();
        //
        // 数据分流
        //       目前所有的日志数据，全部都耦合在了一个主题中 topic_log
        //        页面访问主题、启动程序主题、报错主题、曝光主题、页面动作主题。
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {
        };
        OutputTag<JSONObject> startTag = new OutputTag<JSONObject>("start") {
        };
        OutputTag<JSONObject> actionTag = new OutputTag<JSONObject>("action") {
        };
        OutputTag<JSONObject> errorTag = new OutputTag<JSONObject>("error") {
        };


        //   error :  有err元素的
        // 无error
        //   start :  有 start 元素
        //   page : 有page元素    去除掉displays 和 actions 元素
        //   display : 有displays 元素    1条 拆多条
        //   action : 有actions元素       1 条 拆多条
        SingleOutputStreamOperator<JSONObject> pageStream = jsonObjAfterIsnewCheckStream.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject jsonObject, Context ctx, Collector<JSONObject> out) throws Exception {
                //   error :  有err 元素的
                JSONObject errJsonObj = jsonObject.getJSONObject("err");
                if (errJsonObj != null && errJsonObj.size() > 0) {
                    ctx.output(errorTag, jsonObject);
                } else {
                    // 无error
                    //   start :  有 start 元素
                    JSONObject startJsonObj = jsonObject.getJSONObject("start");
                    if (startJsonObj != null && startJsonObj.size() > 0) {
                        ctx.output(startTag, jsonObject);
                    } else {
                        // 非启动  // 必进page  有可能进 display action

                        //   display : 有displays 元素    1条 拆多条

                        JSONArray displaysJsonArray = jsonObject.getJSONArray("displays");
                        if (displaysJsonArray != null && displaysJsonArray.size() > 0) {
                            for (int i = 0; i < displaysJsonArray.size(); i++) {
                                JSONObject displayJsonObj = displaysJsonArray.getJSONObject(i);
                                JSONObject diplayAllJsonObj = new JSONObject();
                                diplayAllJsonObj.put("display", displayJsonObj);
                                diplayAllJsonObj.put("common", jsonObject.getJSONObject("common"));
                                diplayAllJsonObj.put("page", jsonObject.getJSONObject("page"));
                                diplayAllJsonObj.put("ts", jsonObject.getLong("ts"));
                                ctx.output(displayTag, diplayAllJsonObj);
                            }
                        }
                        //   action : 有actions元素       1 条 拆多条
                        JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");
                        if (actionsJsonArray != null && actionsJsonArray.size() > 0) {
                            for (int i = 0; i < actionsJsonArray.size(); i++) {
                                JSONObject actionJsonObj = actionsJsonArray.getJSONObject(i);
                                JSONObject actionAllJsonObj = new JSONObject();
                                actionAllJsonObj.put("action", actionJsonObj);
                                actionAllJsonObj.put("common", jsonObject.getJSONObject("common"));
                                actionAllJsonObj.put("page", jsonObject.getJSONObject("page"));
                                actionAllJsonObj.put("page_ts", jsonObject.getLong("ts"));
                                ctx.output(actionTag, actionAllJsonObj);
                            }
                        }
                        //   page : 有page元素    去除掉displays 和 actions 元素
                        jsonObject.remove("actions");   // 如果需要分析action 或者displays 直接取对应的主题即可
                        jsonObject.remove("displays");  // 没有必要再从page中分析action 或者displays  所以删除
                        out.collect(jsonObject); //进主流
                    }

                }


            }
        });


        DataStream<String> displayStream = pageStream.getSideOutput(displayTag).map(jsonObj -> jsonObj.toJSONString());
        DataStream<String> actionStream = pageStream.getSideOutput(actionTag).map(jsonObj -> jsonObj.toJSONString());
        DataStream<String> startStream = pageStream.getSideOutput(startTag).map(jsonObj -> jsonObj.toJSONString());
        DataStream<String> errorStream = pageStream.getSideOutput(errorTag).map(jsonObj -> jsonObj.toJSONString());
        SingleOutputStreamOperator<String> pageStrStream = pageStream.map(jsonObj -> jsonObj.toJSONString());


        displayStream.print("display::::");
        actionStream.print("action::::");
        startStream.print("start::::");
        errorStream.print("error::::");
        pageStream.print("page::::");


        // pageStrStream.print();
        displayStream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_display_log"));
        actionStream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_action_log"));
        startStream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_start_log"));
        errorStream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_error_log"));
        pageStrStream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_page_log"));

        dirtyDataStream.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_dirty"));

        //4   env的执行
        env.execute();

    }
}
